package rdd

import org.apache.spark.{SparkConf, SparkContext}

object RDD_Create {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("RDD_Create")

    val sc = new SparkContext(conf)

    // 可以从列表中创建RDD
    val dataList = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val rdd1 = sc.parallelize(dataList,4)
    println(rdd1.getNumPartitions)

    // 可以从集合（范围Range）中创建RDD
    val dataList2 = List.range(1, 11, 2)
    val rdd2 = sc.parallelize(dataList2, 6)
    println(rdd2.getNumPartitions)

    // 可以从集合（数组Array）中创建RDD
    val dataArray = Array(2, 4, 6, 8, 10)
    val rdd3 = sc.parallelize(dataArray, 8)
    println(rdd3.getNumPartitions)
    val dataArray2 = Array("明月几时有", "把酒问青天", "不知天上宫阙", "今夕是何年")
    val strRDD = sc.parallelize(dataArray2,6)
    // 查看RDD上的数据（注：这个操作是在executor端执行的）
    strRDD.foreach(str => println(str))

    // 从存储系统读取数据集创建RDD
    val fileRDD = sc.textFile("data/word.txt")
    /*
    fileRDD
      .flatMap(line => line.split("\\s+"))
      .map(word => (word,1))
      .reduceByKey((x,y) => x+y)
      .foreach(println)
     */

    // 从已有的RDD转换得到新的RDD
    // 现有RDD是 fileRDD
    val mapRDD = fileRDD.map(str => str.toUpperCase)  // 转换得到的RDD
    mapRDD.foreach(println)

  }
}
